package com.shujia.datahub;

import com.aliyun.datahub.client.DatahubClient;
import com.aliyun.datahub.client.DatahubClientBuilder;
import com.aliyun.datahub.client.auth.AliyunAccount;
import com.aliyun.datahub.client.common.DatahubConfig;
import com.aliyun.datahub.client.exception.DatahubClientException;
import com.aliyun.datahub.client.http.HttpConfig;
import com.aliyun.datahub.client.model.PutRecordsResult;
import com.aliyun.datahub.client.model.RecordEntry;
import com.aliyun.datahub.client.model.RecordSchema;
import com.aliyun.datahub.client.model.TupleRecordData;

import java.util.ArrayList;
import java.util.List;

public class DatahubProducer {
    // 写入Tuple型数据
    public static void tupleExample(String project, String topic, int retryTimes) {
        // Endpoint以Region: 华东1为例，其他Region请按实际情况填写
        String endpoint = "https://dh-cn-shanghai.aliyuncs.com";
        String accessId = "LTAIvXl3QTWkqvYc";
        String accessKey = "LUkTK7tY2meir3XuPfSa8RKmbuLQGU";

        // 创建DataHubClient实例
        DatahubClient datahubClient = DatahubClientBuilder.newBuilder()
                .setDatahubConfig(
                        new DatahubConfig(endpoint,
                                // 是否开启二进制传输，服务端2.12版本开始支持
                                new AliyunAccount(accessId, accessKey), true))
                //专有云使用出错尝试将参数设置为           false
                // HttpConfig可不设置，不设置时采用默认值
                .setHttpConfig(new HttpConfig()
                        .setCompressType(HttpConfig.CompressType.LZ4) // 读写数据推荐打开网络传输 LZ4压缩
                        .setConnTimeout(10000))
                .build();

        // 获取schema
        RecordSchema recordSchema = datahubClient.getTopic(project, topic).getRecordSchema();

        // 使用for循环构建十条数据
        List<RecordEntry> recordEntries = new ArrayList<>();

        for (int i = 0; i < 10; ++i) {
            RecordEntry recordEntry = new RecordEntry();
            // 对每条数据设置额外属性，例如ip 机器名等。可以不设置额外属性，不影响数据写入

            TupleRecordData data = new TupleRecordData(recordSchema);

            data.setField("id", "0000000" + i);
            data.setField("name", "zhangSan" + i);
            data.setField("age", "18" + i);
            recordEntry.setRecordData(data);

            recordEntries.add(recordEntry);
        }

        // 往Datahub中写构建好的数据
        try {
            // 往datahub 写数据
            PutRecordsResult result = datahubClient.putRecords(project, topic, recordEntries);
            // 获取失败的条数
            int i = result.getFailedRecordCount();
            if (i > 0) {
                retry(datahubClient, result.getFailedRecords(), retryTimes, project, topic);
            }
        } catch (DatahubClientException e) {
            System.out.println("requestId:" + e.getRequestId() + "\tmessage:" + e.getErrorMessage());
        }
    }

    //重试机制
    public static void retry(DatahubClient client, List<RecordEntry> records, int retryTimes, String project, String topic) {
        boolean suc = false;
        while (retryTimes != 0) {
            retryTimes = retryTimes - 1;
            PutRecordsResult recordsResult = client.putRecords(project, topic, records);
            if (recordsResult.getFailedRecordCount() > 0) {
                retry(client, recordsResult.getFailedRecords(), retryTimes, project, topic);
            }
            suc = true;
            break;
        }
        if (!suc) {
            System.out.println("retryFailure");
        }
    }

    public static void main(String[] args) {
        DatahubProducer.tupleExample("sentiment", "test1", 3);

    }
}
